// #pragma once
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include "Channel.hpp"
#include "Work.hpp"
using namespace std;
namespace mq
{
    class Connection
    {
    public:
        using ConnectionPtr = std::shared_ptr<Connection>;
        Connection(const string &ip, const int port, const AsyncWorker::AsyncWorkerPtr &worker)
            : _latch(1),
              _client(_loopthread.startLoop(), muduo::net::InetAddress(ip, port), "Client"),
              _dispatcher(bind(&Connection::onUnknownMessage, this, placeholders::_1, placeholders::_2, placeholders::_3)),
              _codec(make_shared<ProtobufCodec>(bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, placeholders::_1, placeholders::_2, placeholders::_3))),
              _channel_manager(make_shared<ChannelManager>()), _worker(worker)
        {
            _dispatcher.registerMessageCallback<basicCommonResponse>(bind(&Connection::basicResponse, this, placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(bind(&Connection::consumeResponse, this, placeholders::_1, placeholders::_2, placeholders::_3));

            _client.setConnectionCallback(bind(&Connection::onConnection, this, placeholders::_1));
            _client.setMessageCallback(bind(&ProtobufCodec::onMessage, _codec.get(), placeholders::_1, placeholders::_2, placeholders::_3));
        }
        void connect()
        {
            _client.connect();
            _latch.wait();
        }
        Channel::ChannelPtr openChannel()
        {
            Channel::ChannelPtr cp = _channel_manager->create(_conn, _codec);
            if (cp.get() == nullptr)
            {
                DLOG("打开信道失败");
                return Channel::ChannelPtr();
            }
            bool ret = cp->openChannel();
            if (ret == false)
            {
                DLOG("打开信道失败");
                return Channel::ChannelPtr();
            }
            return cp;
        }
        void closeChannel(const Channel::ChannelPtr channel)
        {
            channel->closeChannel();
            _channel_manager->remove(channel->cid());
        }

    private:
        void basicResponse(const muduo::net::TcpConnectionPtr &conn, const basicCommonResponsePtr &message, muduo::Timestamp)
        {
            Channel::ChannelPtr cp = _channel_manager->get(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("不存在目标信道: %s", message->cid().c_str());
                return;
            }
            cp->putBasicResponse(message);
        }
        void consumeResponse(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr &message, muduo::Timestamp)
        {
            Channel::ChannelPtr cp = _channel_manager->get(message->cid());
            if (cp.get() == nullptr)
            {
                DLOG("不存在目标信道: %s", message->cid().c_str());
                return;
            }
            _worker->_pool.push([cp, message]()
                                { cp->consume(message); });
        }
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            DLOG("onUnknownMessage: %s", message->GetTypeName().c_str());
            conn->shutdown();
        }
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected() == true)
            {
                _latch.countDown();
                _conn = conn;
            }
            else
            {
                _conn.reset();
            }
        }

    private:
        muduo::CountDownLatch _latch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::TcpClient _client;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        ChannelManager::ChannelManagerPtr _channel_manager;
        AsyncWorker::AsyncWorkerPtr _worker;
    };
}